package com.bw.gmall.realtime.app.dws;


import java.sql.Connection;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.UserProfileStats;
import com.bw.gmall.realtime.common.GmallConfig;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.DimUtil;
import com.bw.gmall.realtime.utils.DruidDSUtil;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.alibaba.druid.pool.DruidPooledConnection;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;

/**
 * 人群特征统计应用
 * 统计店铺人群的性别、年龄、城市分布、淘气值分布等特征
 * - UserProfileStatsApp.java ：用户画像特征统计
 */
/**
 * 用户域用户画像统计应用
 * 负责统计用户画像特征数据，包括性别、年龄、地域等维度的用户分布
 * 主要功能点：
 * - 从dwd_traffic_page_log主题读取数据
 * - 涉及用户画像特征统计（性别、年龄等）
 * - 使用Druid连接池进行数据管理
 * - 调用维度查询工具DimUtil进行维度信息关联
 * - 输出用户画像统计结果
 * 
 * 数据流向：
 * dwd层页面日志数据 -> UserProfileStatsApp -> 处理后输出用户画像统计结果
 */
public class UserProfileStatsApp {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 读取Kafka的页面日志主题
        String pageTopic = "dwd_traffic_page_log";
        String groupId = "user_profile_stats_group";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId);
        SingleOutputStreamOperator<JSONObject> jsonObjStream = env.addSource(kafkaSource)
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        return JSON.parseObject(s);
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>
                                forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts");
                                    }
                                })
                );

        // 3. 关联用户维度数据，提取人群特征
        SingleOutputStreamOperator<UserProfileStats> userProfileStream = jsonObjStream
                .flatMap(new FlatMapFunction<JSONObject, UserProfileStats>() {
                    @Override
                    public void flatMap(JSONObject jsonObject, Collector<UserProfileStats> out) throws Exception {
                        // 获取统计时间和类型
                        String statsTime = DateFormatUtil.toYmdHms(jsonObject.getLong("ts"));
                        String statsType = "10s";

                        // 获取用户ID
                        String userId = jsonObject.containsKey("user_id") ? jsonObject.getString("user_id") : "";
                        String mid = jsonObject.getString("mid");

                        // 判断人群类型
                        String userGroupType = getUserGroupType(jsonObject);

                        // 从Redis获取用户维度信息（注释掉Phoenix查询，因为Phoenix数据库没有维度表）
                        JSONObject userDim = null;
                        // 直接标记用户维度信息为空，避免连接Phoenix数据库
                        if (!userId.isEmpty()) {
                            System.out.println("Phoenix数据库无维度表，跳过维度信息查询，用户ID: " + userId);
                        }

                        // 创建基础统计项 - 总用户计数
                        UserProfileStats totalStats = new UserProfileStats();
                        totalStats.setStatsTime(statsTime);
                        totalStats.setStatsType(statsType);
                        totalStats.setUserGroupType(userGroupType);
                        totalStats.setFeatureType("basic");
                        totalStats.setFeatureValue("total"); // 总用户计数
                        totalStats.setUserCount(1L);
                        out.collect(totalStats);
                        
                        // 创建用户类型统计项
                        String userType = userId.isEmpty() ? "anonymous" : "registered";
                        UserProfileStats typeStats = new UserProfileStats();
                        typeStats.setStatsTime(statsTime);
                        typeStats.setStatsType(statsType);
                        typeStats.setUserGroupType(userGroupType);
                        typeStats.setFeatureType("user_type");
                        typeStats.setFeatureValue(userType);
                        typeStats.setUserCount(1L);
                        out.collect(typeStats);

                        // 如果有用户信息，则进行详细分析
                        if (userDim != null) {
                            // 性别分布
                            String gender = userDim.getString("gender");
                            if (gender != null) {
                                UserProfileStats genderStats = new UserProfileStats();
                                genderStats.setStatsTime(statsTime);
                                genderStats.setStatsType(statsType);
                                genderStats.setUserGroupType(userGroupType);
                                genderStats.setFeatureType("gender");
                                genderStats.setFeatureValue(gender);
                                genderStats.setUserCount(1L);
                                out.collect(genderStats);
                            }

                            // 年龄分布
                            String birthday = userDim.getString("birthday");
                            if (birthday != null && !birthday.isEmpty()) {
                                int age = calculateAge(birthday);
                                String ageRange = getAgeRange(age);
                                UserProfileStats ageStats = new UserProfileStats();
                                ageStats.setStatsTime(statsTime);
                                ageStats.setStatsType(statsType);
                                ageStats.setUserGroupType(userGroupType);
                                ageStats.setFeatureType("age");
                                ageStats.setFeatureValue(ageRange);
                                ageStats.setUserCount(1L);
                                out.collect(ageStats);
                            }

                            // 城市分布
                            String city = userDim.getString("city");
                            if (city != null) {
                                UserProfileStats cityStats = new UserProfileStats();
                                cityStats.setStatsTime(statsTime);
                                cityStats.setStatsType(statsType);
                                cityStats.setUserGroupType(userGroupType);
                                cityStats.setFeatureType("city");
                                cityStats.setFeatureValue(city);
                                cityStats.setUserCount(1L);
                                out.collect(cityStats);
                            }

                            // 淘气值分布
                            String naughtyScore = userDim.containsKey("naughty_score") ?
                                            userDim.getString("naughty_score") : "";
                            if (!naughtyScore.isEmpty()) {
                                try {
                                    int score = Integer.parseInt(naughtyScore);
                                    String scoreRange = getNaughtyScoreRange(score);
                                    UserProfileStats scoreStats = new UserProfileStats();
                                    scoreStats.setStatsTime(statsTime);
                                    scoreStats.setStatsType(statsType);
                                    scoreStats.setUserGroupType(userGroupType);
                                    scoreStats.setFeatureType("naughty_score");
                                    scoreStats.setFeatureValue(scoreRange);
                                    scoreStats.setUserCount(1L);
                                    out.collect(scoreStats);
                                } catch (NumberFormatException e) {
                                    System.err.println("解析淘气值失败: " + naughtyScore + ", 错误信息: " + e.getMessage());
                                }
                            }
                        } else if (userId.isEmpty()) {
                            // 只对真正的匿名用户添加anonymous特征统计
                            UserProfileStats anonymousStats = new UserProfileStats();
                            anonymousStats.setStatsTime(statsTime);
                            anonymousStats.setStatsType(statsType);
                            anonymousStats.setUserGroupType(userGroupType);
                            anonymousStats.setFeatureType("anonymous");
                            anonymousStats.setFeatureValue("true");
                            anonymousStats.setUserCount(1L);
                            out.collect(anonymousStats);
                        }
                    }
                });

        // 4. 聚合统计各类特征
        SingleOutputStreamOperator<UserProfileStats> aggregatedStream = userProfileStream
                .keyBy(new KeySelector<UserProfileStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(UserProfileStats stats) throws Exception {
                        // 使用单独的字段而不是拼接，确保正确分组
                        return new Tuple4<>(stats.getStatsTime(),
                                           stats.getStatsType(),
                                           stats.getUserGroupType(),
                                           stats.getFeatureType() + ":" + stats.getFeatureValue());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .sum("userCount");

        // 5. 计算各类特征的占比
        SingleOutputStreamOperator<UserProfileStats> finalStatsStream = aggregatedStream
                .keyBy(new KeySelector<UserProfileStats, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(UserProfileStats stats) throws Exception {
                        // 只按统计时间、统计类型和用户群体类型分组
                        return new Tuple3<>(stats.getStatsTime(),
                                           stats.getStatsType(),
                                           stats.getUserGroupType());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .apply(new WindowFunction<UserProfileStats, UserProfileStats,
                        Tuple3<String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple3<String, String, String> key, TimeWindow window,
                                     Iterable<UserProfileStats> input, Collector<UserProfileStats> out) throws Exception {
                        // 计算总人数（从basic:total统计项获取）
                        long totalUsers = 0;
                        List<UserProfileStats> statsList = new ArrayList<>();

                        for (UserProfileStats stats : input) {
                            statsList.add(stats);
                            // 只从总用户统计项获取总人数
                            if ("basic".equals(stats.getFeatureType()) && "total".equals(stats.getFeatureValue())) {
                                totalUsers = stats.getUserCount();
                            }
                        }

                        // 计算各类特征的占比
                        for (UserProfileStats stats : statsList) {
                            // 跳过总用户统计项的占比计算
                            if (!("basic".equals(stats.getFeatureType()) && "total".equals(stats.getFeatureValue()))) {
                                double proportion = totalUsers > 0 ? (double) stats.getUserCount() / totalUsers : 0.0;
                                stats.setProportion(proportion);
                            }
                            stats.setCreateTime(new Date());
                            out.collect(stats);
                        }
                    }
                });

        // 6. 将结果写入ClickHouse
        finalStatsStream.addSink(MyClickHouseUtil.getSinkFunction(
                "insert into dws_user_profile_stats_window values(?,?,?,?,?,?,?,?)"
        ));

        // 7. 执行作业
        env.execute("UserProfileStatsApp");
    }

    /**
     * 判断用户群体类型
     * 根据用户行为判断是进店人群、商品访问人群还是转化人群
     */
    private static String getUserGroupType(JSONObject jsonObject) {
        // 获取页面信息
        JSONObject page = jsonObject.getJSONObject("page");
        String pageId = page.getString("page_id");

        // 获取事件信息
        if (jsonObject.containsKey("actions")) {
            // 检查是否有购买相关行为
            if (hasPurchaseAction(jsonObject)) {
                return "conversion_group";
            }
        }

        // 检查是否访问了商品详情页
        if ("good_detail".equals(pageId)) {
            return "goods_visit_group";
        }

        // 默认是进店人群
        return "store_visit_group";
    }

    /**
     * 检查是否有购买相关行为
     */
    private static boolean hasPurchaseAction(JSONObject jsonObject) {
        // 更全面的购买行为判断
        // 1. 检查是否有订单ID
        if (jsonObject.containsKey("order_id") && jsonObject.getString("order_id") != null && !jsonObject.getString("order_id").isEmpty()) {
            return true;
        }
        
        // 2. 检查是否访问了支付成功页面
        if (jsonObject.containsKey("page") && jsonObject.getJSONObject("page") != null) {
            String pageId = jsonObject.getJSONObject("page").getString("page_id");
            if ("payment_success".equals(pageId) || "order_confirm".equals(pageId)) {
                return true;
            }
        }
        
        // 3. 检查是否有购买相关的actions
        if (jsonObject.containsKey("actions") && jsonObject.getJSONArray("actions") != null) {
            for (int i = 0; i < jsonObject.getJSONArray("actions").size(); i++) {
                JSONObject action = jsonObject.getJSONArray("actions").getJSONObject(i);
                if ("place_order".equals(action.getString("action_id")) || "pay_order".equals(action.getString("action_id"))) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * 根据生日计算年龄
     */
    private static int calculateAge(String birthday) {
        try {
            Calendar birthCal = Calendar.getInstance();
            Calendar nowCal = Calendar.getInstance();

            // 解析生日字符串，假设格式为yyyy-MM-dd
            String[] parts = birthday.split("-");
            if (parts.length == 3) {
                birthCal.set(Calendar.YEAR, Integer.parseInt(parts[0]));
                birthCal.set(Calendar.MONTH, Integer.parseInt(parts[1]) - 1);
                birthCal.set(Calendar.DAY_OF_MONTH, Integer.parseInt(parts[2]));

                int age = nowCal.get(Calendar.YEAR) - birthCal.get(Calendar.YEAR);
                if (nowCal.get(Calendar.DAY_OF_YEAR) < birthCal.get(Calendar.DAY_OF_YEAR)) {
                    age--;
                }
                return age;
            }
        } catch (Exception e) {
            // 记录解析异常日志
            System.err.println("解析生日失败: " + birthday + ", 错误信息: " + e.getMessage());
            e.printStackTrace();
        }
        return 0;
    }

    /**
     * 获取年龄段
     */
    private static String getAgeRange(int age) {
        if (age < 18) return "<18";
        else if (age < 25) return "18-24";
        else if (age < 35) return "25-34";
        else if (age < 45) return "35-44";
        else if (age < 55) return "45-54";
        else return ">=55";
    }

    /**
     * 获取淘气值区间
     */
    private static String getNaughtyScoreRange(int score) {
        if (score < 300) return "<300";
        else if (score < 500) return "300-499";
        else if (score < 700) return "500-699";
        else if (score < 900) return "700-899";
        else return ">=900";
    }
}